package com.hadoop.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderRefund extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderRefund().start(Constant.TOPIC_DB, Constant.TOPIC_DWD_TRADE_ORDER_REFUND, 4, 10017);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId) {
        // 1.1 读取 topic_db
        readOdsDb(tEnv, Constant.TOPIC_DWD_TRADE_ORDER_REFUND);
        // 1.2 读取 字典表
        readBaseDic(tEnv);

        // 2. 过滤退单表数据 order_refund_info   insert
        Table orderRefundInfo = tEnv.sqlQuery(
                "SELECT " +
                        "data['id'] id, " +
                        "data['user_id'] user_id, " +
                        "data['order_id'] order_id, " +
                        "data['sku_id'] sku_id, " +
                        "data['refund_type'] refund_type, " +
                        "data['refund_num'] refund_num, " +
                        "data['refund_amount'] refund_amount, " +
                        "data['refund_reason_type'] refund_reason_type, " +
                        "data['refund_reason_txt'] refund_reason_txt, " +
                        "data['create_time'] create_time, " +
                        "proc_time, " +
                        "ts " +
                        "FROM topic_db " +
                        "WHERE `database` = 'gmall' " +
                        "AND `table` = 'order_refund_info' " +
                        "AND `type` = 'insert' ");
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);

        // 3. 过滤订单表中的退单数据: order_info  update
        Table orderInfo = tEnv.sqlQuery(
                "SELECT " +
                        "data['id'] id, " +
                        "data['province_id'] province_id, " +
                        "`old` " +
                        "FROM topic_db " +
                        "WHERE `database` = 'gmall' " +
                        "AND `table` = 'order_info' " +
                        "AND `type` = 'update' " +
                        "AND `old`['order_status'] IS NOT NULL " +
                        "AND `data`['order_status'] = '1005' ");
        tEnv.createTemporaryView("order_info", orderInfo);

        // 4. join: 普通的和 lookup join
        Table result = tEnv.sqlQuery(
                "SELECT " +
                        "ri.id, " +
                        "ri.user_id, " +
                        "ri.order_id, " +
                        "ri.sku_id, " +
                        "oi.province_id, " +
                        "DATE_FORMAT(ri.create_time, 'yyyy-MM-dd') date_id, " +
                        "ri.create_time, " +
                        "ri.refund_type, " +
                        "dic1.dic_name, " +
                        "ri.refund_reason_type, " +
                        "dic2.dic_name, " +
                        "ri.refund_reason_txt, " +
                        "ri.refund_num, " +
                        "ri.refund_amount, " +
                        "ri.ts " +
                        "FROM order_refund_info ri " +
                        "JOIN order_info oi ON ri.order_id = oi.id " +
                        "JOIN base_dic FOR SYSTEM_TIME AS OF ri.proc_time AS dic1 ON ri.refund_type = dic1.dic_code " +
                        "JOIN base_dic FOR SYSTEM_TIME AS OF ri.proc_time AS dic2 ON ri.refund_reason_type = dic2.dic_code ");

        // 5. 写出到 kafka
        tEnv.executeSql(
                "CREATE TABLE dwd_trade_order_refund (" +
                        "id STRING, " +
                        "user_id STRING, " +
                        "order_id STRING, " +
                        "sku_id STRING, " +
                        "province_id STRING, " +
                        "date_id STRING, " +
                        "create_time STRING, " +
                        "refund_type_code STRING, " +
                        "refund_type_name STRING, " +
                        "refund_reason_type_code STRING, " +
                        "refund_reason_type_name STRING, " +
                        "refund_reason_txt STRING, " +
                        "refund_num STRING, " +
                        "refund_amount STRING, " +
                        "ts BIGINT " +
                        ")" + SQLUtil.getKafkaDDLSink(Constant.TOPIC_DWD_TRADE_ORDER_REFUND));

        result.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_REFUND);
    }
}